{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "bfe458c6-2671-4a9f-a16a-b19a99681b87",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "import os\n",
    "findspark.init()\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "jars_directory = r\"E:\\jupyterlab\\pyspark环境\\spark_graph处理\\jar\"\n",
    "# 获取目录中的所有JAR文件\n",
    "jars = [os.path.join(jars_directory, jar) for jar in os.listdir(jars_directory) if jar.endswith(\".jar\")]\n",
    "\n",
    "# 以逗号分隔的字符串形式传递给 Spark\n",
    "jars_str = \",\".join(jars)\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "        .config(\"spark.jars\", jars_str) \\\n",
    "        .config(\"spark.driver.host\", \"localhost\") \\\n",
    "        .config(\"spark.driver.memory\", \"2g\") \\\n",
    "        .config(\"spark.executor.memory\", \"4g\") \\\n",
    "        .appName('xj_spark') \\\n",
    "        .master(\"local[*]\") \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "bb44b5b5-d944-4c53-9248-d5ee962e273b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+---+\n",
      "| id|   name|age|\n",
      "+---+-------+---+\n",
      "|  a|  Alice| 34|\n",
      "|  b|    Bob| 36|\n",
      "|  c|Charlie| 30|\n",
      "+---+-------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "v = spark.createDataFrame([\n",
    "  (\"a\", \"Alice\", 34),\n",
    "  (\"b\", \"Bob\", 36),\n",
    "  (\"c\", \"Charlie\", 30),\n",
    "], [\"id\", \"name\", \"age\"])\n",
    "v.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "6a4432ae-8662-45b6-a678-ceee3c2bddc3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+------------+\n",
      "|src|dst|relationship|\n",
      "+---+---+------------+\n",
      "|  a|  b|      friend|\n",
      "|  b|  c|      follow|\n",
      "|  c|  b|      follow|\n",
      "+---+---+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "e = spark.createDataFrame([\n",
    "  (\"a\", \"b\", \"friend\"),\n",
    "  (\"b\", \"c\", \"follow\"),\n",
    "  (\"c\", \"b\", \"follow\"),\n",
    "], [\"src\", \"dst\", \"relationship\"])\n",
    "\n",
    "e.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "4c8bd7e3-0a40-44a8-8ef9-3fae813da1a5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "| id|inDegree|\n",
      "+---+--------+\n",
      "|  c|       1|\n",
      "|  b|       2|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from graphframes import *\n",
    "g = GraphFrame(v, e)\n",
    "# 查询：获取每个顶点的度数。入度（即指向该顶点的边的数量）\n",
    "g.inDegrees.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "2a066f90-8476-48e1-8f51-7a15ce67298c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 查询：统计图中“follow”连接的数量。\n",
    "g.edges.filter(\"relationship = 'follow'\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "f7c0bfb3-5332-485f-84f6-8aa2ea35e081",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------------------+\n",
      "| id|          pagerank|\n",
      "+---+------------------+\n",
      "|  b|1.0905890109440908|\n",
      "|  a|              0.01|\n",
      "|  c|1.8994109890559092|\n",
      "+---+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pageRank 简单理解为关系中的权重算法\n",
    "results = g.pageRank(resetProbability=0.01, maxIter=20)\n",
    "results.vertices.select(\"id\", \"pagerank\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aea3ae29-f687-4c8d-a591-b98beb02a7c5",
   "metadata": {},
   "source": [
    "### 以下示例演示了如何从顶点和边DataFrames创建GraphFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "bf792bc2-5ed0-4122-9b1a-8582d629da5b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Vertex DataFrame\n",
    "v = spark.createDataFrame([\n",
    "  (\"a\", \"Alice\", 34),\n",
    "  (\"b\", \"Bob\", 36),\n",
    "  (\"c\", \"Charlie\", 30),\n",
    "  (\"d\", \"David\", 29),\n",
    "  (\"e\", \"Esther\", 32),\n",
    "  (\"f\", \"Fanny\", 36),\n",
    "  (\"g\", \"Gabby\", 60)\n",
    "], [\"id\", \"name\", \"age\"])\n",
    "# Edge DataFrame\n",
    "e = spark.createDataFrame([\n",
    "  (\"a\", \"b\", \"friend\"),\n",
    "  (\"b\", \"c\", \"follow\"),\n",
    "  (\"c\", \"b\", \"follow\"),\n",
    "  (\"f\", \"c\", \"follow\"),\n",
    "  (\"e\", \"f\", \"follow\"),\n",
    "  (\"e\", \"d\", \"friend\"),\n",
    "  (\"d\", \"a\", \"friend\"),\n",
    "  (\"a\", \"e\", \"friend\")\n",
    "], [\"src\", \"dst\", \"relationship\"])\n",
    "# Create a GraphFrame\n",
    "g = GraphFrame(v, e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9a7f28fa-4c91-41ec-882c-667b12901f26",
   "metadata": {},
   "outputs": [],
   "source": [
    " # 报错 原因  https://github.com/graphframes/graphframes/issues/414\n",
    "from graphframes.examples import Graphs\n",
    "g = Graphs.friends()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "c326f8db-9eb7-4412-8fd8-ebdefd207be1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+---+\n",
      "| id|   name|age|\n",
      "+---+-------+---+\n",
      "|  a|  Alice| 34|\n",
      "|  b|    Bob| 36|\n",
      "|  c|Charlie| 30|\n",
      "|  d|  David| 29|\n",
      "|  e| Esther| 32|\n",
      "|  f|  Fanny| 36|\n",
      "|  g|  Gabby| 60|\n",
      "+---+-------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 显示顶点和边数据 DataFrame\n",
    "g.vertices.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "6f1ef4ae-af7a-4a92-a4d3-218a2749f250",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+------------+\n",
      "|src|dst|relationship|\n",
      "+---+---+------------+\n",
      "|  a|  b|      friend|\n",
      "|  b|  c|      follow|\n",
      "|  c|  b|      follow|\n",
      "|  f|  c|      follow|\n",
      "|  e|  f|      follow|\n",
      "|  e|  d|      friend|\n",
      "|  d|  a|      friend|\n",
      "|  a|  e|      friend|\n",
      "+---+---+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "g.edges.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "2d807b1a-e0f9-44e3-9cf9-99473260ca70",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "| id|inDegree|\n",
      "+---+--------+\n",
      "|  f|       1|\n",
      "|  e|       1|\n",
      "|  d|       1|\n",
      "|  c|       2|\n",
      "|  b|       2|\n",
      "|  a|       1|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 获取包含列“id”和“inDegree”（以度为单位）的DataFrame\n",
    "vertexInDegrees = g.inDegrees\n",
    "vertexInDegrees.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "b617957e-dcde-483f-93f7-9b8f3f6b5df0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|min(age)|\n",
      "+--------+\n",
      "|      29|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 在图表中找出最年轻用户的年龄。这将查询顶点DataFrame。\n",
    "g.vertices.groupBy().min(\"age\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "c2ccea00-4493-42dc-a20c-96e902e476fd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "g.edges.filter(\"relationship = 'follow'\").count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "43f4dac9-8a9d-4e1f-9f54-d1e886f78813",
   "metadata": {},
   "source": [
    "1、顶点用括号（a）表示，而边用方括号[e]表示。\n",
    "2、Motifs 不允许包含没有任何命名元素的边：“（）-[]->（）”和“！（）-[]->（）”是禁止使用的术语。\n",
    "3、Motifs 不允许在否定项中包含命名边（因为这些命名边永远不会出现在结果中）。例如，“！（a）-[ab]->（b）”无效，但“！（b）-[]->（a）”有效。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "776da05c-4459-43ee-ad6d-531524dc0847",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------+----------------+--------------+\n",
      "|               a|             e|               b|            e2|\n",
      "+----------------+--------------+----------------+--------------+\n",
      "|{c, Charlie, 30}|{c, b, follow}|    {b, Bob, 36}|{b, c, follow}|\n",
      "|    {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|{c, b, follow}|\n",
      "+----------------+--------------+----------------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "motifs = g.find(\"(a)-[e]->(b); (b)-[e2]->(a)\")\n",
    "motifs.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "86eb1bcf-3cd6-4911-9514-ae30c88eafba",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------+------------+--------------+\n",
      "|               a|             e|           b|            e2|\n",
      "+----------------+--------------+------------+--------------+\n",
      "|{c, Charlie, 30}|{c, b, follow}|{b, Bob, 36}|{b, c, follow}|\n",
      "+----------------+--------------+------------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "motifs.filter(\"b.age > 30\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "aa929fc5-8018-4d68-b3a6-164732a389d2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------+----------------+--------------+----------------+--------------+----------------+\n",
      "|               a|            ab|               b|            bc|               c|            cd|               d|\n",
      "+----------------+--------------+----------------+--------------+----------------+--------------+----------------+\n",
      "|  {d, David, 29}|{d, a, friend}|  {a, Alice, 34}|{a, e, friend}| {e, Esther, 32}|{e, f, follow}|  {f, Fanny, 36}|\n",
      "| {e, Esther, 32}|{e, d, friend}|  {d, David, 29}|{d, a, friend}|  {a, Alice, 34}|{a, e, friend}| {e, Esther, 32}|\n",
      "|  {d, David, 29}|{d, a, friend}|  {a, Alice, 34}|{a, e, friend}| {e, Esther, 32}|{e, d, friend}|  {d, David, 29}|\n",
      "|  {a, Alice, 34}|{a, e, friend}| {e, Esther, 32}|{e, f, follow}|  {f, Fanny, 36}|{f, c, follow}|{c, Charlie, 30}|\n",
      "|  {f, Fanny, 36}|{f, c, follow}|{c, Charlie, 30}|{c, b, follow}|    {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|\n",
      "|    {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|{c, b, follow}|    {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|\n",
      "|  {d, David, 29}|{d, a, friend}|  {a, Alice, 34}|{a, b, friend}|    {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|\n",
      "| {e, Esther, 32}|{e, f, follow}|  {f, Fanny, 36}|{f, c, follow}|{c, Charlie, 30}|{c, b, follow}|    {b, Bob, 36}|\n",
      "|{c, Charlie, 30}|{c, b, follow}|    {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|{c, b, follow}|    {b, Bob, 36}|\n",
      "|  {a, Alice, 34}|{a, b, friend}|    {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|{c, b, follow}|    {b, Bob, 36}|\n",
      "| {e, Esther, 32}|{e, d, friend}|  {d, David, 29}|{d, a, friend}|  {a, Alice, 34}|{a, b, friend}|    {b, Bob, 36}|\n",
      "|  {a, Alice, 34}|{a, e, friend}| {e, Esther, 32}|{e, d, friend}|  {d, David, 29}|{d, a, friend}|  {a, Alice, 34}|\n",
      "+----------------+--------------+----------------+--------------+----------------+--------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "chain4 = g.find(\"(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d)\")\n",
    "chain4.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "540b47b1-ba6d-4e2f-87e8-3d3144088bcd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Column<'CASE WHEN (cd[relationship] = friend) THEN (CASE WHEN (bc[relationship] = friend) THEN (CASE WHEN (ab[relationship] = friend) THEN (0 + 1) ELSE 0 END + 1) ELSE CASE WHEN (ab[relationship] = friend) THEN (0 + 1) ELSE 0 END END + 1) ELSE CASE WHEN (bc[relationship] = friend) THEN (CASE WHEN (ab[relationship] = friend) THEN (0 + 1) ELSE 0 END + 1) ELSE CASE WHEN (ab[relationship] = friend) THEN (0 + 1) ELSE 0 END END END'>"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from functools import reduce\n",
    "from pyspark.sql.functions import col, lit, when\n",
    "from pyspark.sql.types import IntegerType\n",
    "sumFriends = lambda cnt,relationship: when(relationship == \"friend\", cnt+1).otherwise(cnt)\n",
    "condition = reduce(lambda cnt,e: sumFriends(cnt, col(e).relationship), [\"ab\", \"bc\", \"cd\"], lit(0))\n",
    "\n",
    "condition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "8a841ffa-5d65-49b8-8def-9c59f7107da9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+--------------+---------------+--------------+---------------+--------------+----------------+\n",
      "|              a|            ab|              b|            bc|              c|            cd|               d|\n",
      "+---------------+--------------+---------------+--------------+---------------+--------------+----------------+\n",
      "| {d, David, 29}|{d, a, friend}| {a, Alice, 34}|{a, e, friend}|{e, Esther, 32}|{e, f, follow}|  {f, Fanny, 36}|\n",
      "|{e, Esther, 32}|{e, d, friend}| {d, David, 29}|{d, a, friend}| {a, Alice, 34}|{a, e, friend}| {e, Esther, 32}|\n",
      "| {d, David, 29}|{d, a, friend}| {a, Alice, 34}|{a, e, friend}|{e, Esther, 32}|{e, d, friend}|  {d, David, 29}|\n",
      "| {d, David, 29}|{d, a, friend}| {a, Alice, 34}|{a, b, friend}|   {b, Bob, 36}|{b, c, follow}|{c, Charlie, 30}|\n",
      "|{e, Esther, 32}|{e, d, friend}| {d, David, 29}|{d, a, friend}| {a, Alice, 34}|{a, b, friend}|    {b, Bob, 36}|\n",
      "| {a, Alice, 34}|{a, e, friend}|{e, Esther, 32}|{e, d, friend}| {d, David, 29}|{d, a, friend}|  {a, Alice, 34}|\n",
      "+---------------+--------------+---------------+--------------+---------------+--------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 筛选有两条以上边 为 friend 的数据\n",
    "chainWith2Friends2 = chain4.where(condition >= 2)\n",
    "chainWith2Friends2.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "05fd8463-4759-4e3b-80ff-5d7ef8e1d947",
   "metadata": {},
   "source": [
    "### 子图"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "5397fc36-415e-4885-ad78-2d46bf44c459",
   "metadata": {},
   "outputs": [],
   "source": [
    "g1 = g.filterVertices(\"age > 30\").filterEdges(\"relationship = 'friend'\").dropIsolatedVertices()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "5af7e5d9-f228-430e-97c8-6e76d2827b72",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+---+\n",
      "| id|   name|age|\n",
      "+---+-------+---+\n",
      "|  a|  Alice| 34|\n",
      "|  b|    Bob| 36|\n",
      "|  c|Charlie| 30|\n",
      "|  d|  David| 29|\n",
      "|  e| Esther| 32|\n",
      "|  f|  Fanny| 36|\n",
      "|  g|  Gabby| 60|\n",
      "+---+-------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "paths = g.find(\"(a)-[e]->(b)\")\\\n",
    "  .filter(\"e.relationship = 'follow'\")\\\n",
    "  .filter(\"a.age < b.age\")\n",
    "\n",
    "e2 = paths.select(\"e.src\", \"e.dst\", \"e.relationship\")\n",
    "\n",
    "g2 = GraphFrame(g.vertices, e2)\n",
    "\n",
    "g2.vertices.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "893fde6a-3f70-4392-8a62-1f1d56cc883d",
   "metadata": {},
   "source": [
    "### 广度优先搜索（BFS）找到从一个顶点（或一组顶点）到另一个顶点的最短路径。开始顶点和结束顶点指定为Spark DataFrame表达式。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "43d69505-67f1-4ade-be63-17a21c575911",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+--------------+--------------+\n",
      "|           from|            e0|            to|\n",
      "+---------------+--------------+--------------+\n",
      "|{e, Esther, 32}|{e, d, friend}|{d, David, 29}|\n",
      "+---------------+--------------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "paths = g.bfs(\"name = 'Esther'\", \"age < 32\")\n",
    "paths.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "cb866b32-d3a6-4ffb-899d-713d750e333d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+--------------+--------------+--------------+----------------+\n",
      "|           from|            e0|            v1|            e1|              to|\n",
      "+---------------+--------------+--------------+--------------+----------------+\n",
      "|{e, Esther, 32}|{e, f, follow}|{f, Fanny, 36}|{f, c, follow}|{c, Charlie, 30}|\n",
      "+---------------+--------------+--------------+--------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "g.bfs(\"name = 'Esther'\", \"age < 32\",edgeFilter=\"relationship != 'friend'\", maxPathLength=3).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "954ab46b-b6e3-408f-a4b7-7f2d0d11b5ee",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------------+\n",
      "| id|   component|\n",
      "+---+------------+\n",
      "|  g|146028888064|\n",
      "|  c|412316860416|\n",
      "|  b|412316860416|\n",
      "|  f|412316860416|\n",
      "|  a|412316860416|\n",
      "|  d|412316860416|\n",
      "|  e|412316860416|\n",
      "+---+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 按照连接分组\n",
    "spark.sparkContext.setCheckpointDir(r\"D:\\spark_local\\graph\\20240801_3\")\n",
    "# 在无向图中，如果两个顶点之间存在一条路径，则它们是连通的。无向图的一个连通分量是该图的一个最大连通子图，即该子图中任意两个顶点都是连通的，且该子图不是其他连通子图的真子集\n",
    "result = g.connectedComponents()\n",
    "result.select(\"id\", \"component\").orderBy(\"component\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "81fff9e4-92bd-4b2c-bb6f-6ade63a8b933",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 在有向图中，如果两个顶点u和v之间既有从u到v的路径，也有从v到u的路径，则称这两个顶点是强连通的。一个有向图的极大强连通子图被称为该图的强连通分量。在强连通分量中，任意两个顶点都是相互可达的\n",
    "result = g.stronglyConnectedComponents(maxIter=10)\n",
    "result.select(\"id\", \"component\").orderBy(\"component\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bca5223c-df82-4a85-84a0-a6526f983cf2",
   "metadata": {},
   "source": [
    "### 标签传播算法（LPA）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "aa99a7e7-de21-48e7-9f67-ea4cac521edd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------------+\n",
      "| id|        label|\n",
      "+---+-------------+\n",
      "|  g| 146028888064|\n",
      "|  b|1047972020224|\n",
      "|  e|1382979469312|\n",
      "|  a|1382979469312|\n",
      "|  f|1460288880640|\n",
      "|  d|1460288880640|\n",
      "|  c|1382979469312|\n",
      "+---+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result = g.labelPropagation(maxIter=5)\n",
    "result.select(\"id\", \"label\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aa47b4dc-e780-43d6-9648-3b7ebdde3f72",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 一些权重的算法\n",
    "results = g.pageRank(resetProbability=0.15, tol=0.01)\n",
    "results.vertices.select(\"id\", \"pagerank\").show()\n",
    "results.edges.select(\"src\", \"dst\", \"weight\").show()\n",
    "results2 = g.pageRank(resetProbability=0.15, maxIter=10)\n",
    "results3 = g.pageRank(resetProbability=0.15, maxIter=10, sourceId=\"a\")\n",
    "results4 = g.parallelPersonalizedPageRank(resetProbability=0.15, sourceIds=[\"a\", \"b\", \"c\", \"d\"], maxIter=10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "054d3601-38ff-409a-aa23-f89ddaac12da",
   "metadata": {},
   "source": [
    "### 最短路径"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "id": "40ced11c-6d7a-45c4-b225-2f42d451ce9b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------------+\n",
      "| id|       distances|\n",
      "+---+----------------+\n",
      "|  g|              {}|\n",
      "|  b|              {}|\n",
      "|  e|{a -> 2, d -> 1}|\n",
      "|  a|{a -> 0, d -> 2}|\n",
      "|  f|              {}|\n",
      "|  d|{a -> 1, d -> 0}|\n",
      "|  c|              {}|\n",
      "+---+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "results = g.shortestPaths(landmarks=[\"a\", \"d\"])\n",
    "results.select(\"id\", \"distances\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "7ffe4992-2baa-4cba-89b2-aaf3497b7f95",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+\n",
      "| id|count|\n",
      "+---+-----+\n",
      "|  g|    0|\n",
      "|  f|    0|\n",
      "|  e|    1|\n",
      "|  d|    1|\n",
      "|  c|    0|\n",
      "|  b|    0|\n",
      "|  a|    1|\n",
      "+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Triangle Count（三角形计数）是用来确定图中每个节点与其1-hop（即直接相连的）节点能够直接形成三角形的个数的一个概念。\n",
    "# 具体来说，三角形是由三个节点组成的集合，其中每个节点都与其他两个节点有直接相连的关系1。\n",
    "results = g.triangleCount()\n",
    "results.select(\"id\", \"count\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "760fe806-1316-4dea-94d4-3d5c8d49f78a",
   "metadata": {},
   "outputs": [],
   "source": [
    "g.vertices.write.parquet(r\"D:\\spark_local\\graph\\input\\20240801_v\")\n",
    "g.edges.write.parquet(r\"D:\\spark_local\\graph\\input\\20240801_e\")\n",
    "\n",
    "sameV = spark.read.parquet(r\"D:\\spark_local\\graph\\input\\20240801_v\")\n",
    "sameE = spark.read.parquet(r\"D:\\spark_local\\graph\\input\\20240801_e\")\n",
    "\n",
    "sameG = GraphFrame(sameV, sameE)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "id": "59ec209e-57cf-4460-af0b-585c9d813bfb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+\n",
      "| id|summedAges|\n",
      "+---+----------+\n",
      "|  f|        62|\n",
      "|  e|        99|\n",
      "|  d|        66|\n",
      "|  c|       108|\n",
      "|  b|        94|\n",
      "|  a|        97|\n",
      "+---+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import sum as sqlsum\n",
    "from graphframes.lib import AggregateMessages as AM\n",
    "msgToSrc = AM.dst[\"age\"]\n",
    "msgToDst = AM.src[\"age\"]\n",
    "agg = g.aggregateMessages(\n",
    "    sqlsum(AM.msg).alias(\"summedAges\"),\n",
    "    sendToSrc=msgToSrc,\n",
    "    sendToDst=msgToDst)\n",
    "agg.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd03c0db-dfc8-4f59-82ab-85c165465d22",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
